package yatan.cluster.job.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import yatan.cluster.job.Job;
import yatan.cluster.job.JobDispatcher;
import yatan.cluster.job.JobDispatcherException;
import yatan.cluster.job.JobResult;

public abstract class AbstractDispatcher implements JobDispatcher {
    private final Logger logger = Logger.getLogger(getClass());

    private final Map<Long, JobStatus> jobStatuses = new HashMap<Long, JobStatus>();
    private final List<JobResult> jobResultCache = new ArrayList<JobResult>();

    private int flushThreashold = 1000;
    private int jobRunningTimeout = 60 * 60 * 1000;

    private int jobErrorLimit = 10;

    private boolean useSeparateThreadFlush = false;

    @Override
    public synchronized ArrayList<Job> dispathJobs(int jobCount) throws JobDispatcherException {
	ArrayList<Job> jobs = new ArrayList<Job>();

	// find jobs from memory
	findJobsFromMemory(jobs, jobCount);

	// if not enough jobs in memory, load some from database
	if (jobs.size() < jobCount) {
	    this.logger.info("Not enough jobs in memory, retrieving new...");
	    for (Job job : retrieveNewJobs()) {
		if (!this.jobStatuses.containsKey(job.getId())) {
		    JobStatus record = new JobStatus();
		    record.setJob(job);
		    this.jobStatuses.put(job.getId(), record);
		}
	    }
	    this.logger.info("Still have unfinished jobs at least " + this.jobStatuses.size());

	    // find jobs from memory
	    findJobsFromMemory(jobs, jobCount);
	}

	this.logger.info("Dispatching " + jobs.size() + " jobs.");
	return jobs;
    }

    @Override
    public synchronized void handleJobError(Job job, Exception error) throws JobDispatcherException {
	this.logger.warn("Job " + job + " exit with error: " + error.getClass().getSimpleName() + "#"
	        + error.getMessage());
	JobStatus jobStatus = this.jobStatuses.get(job.getId());
	if (jobStatus == null) {
	    this.logger.warn("Cannot find the status record with the error job#" + job.getId()
		    + ". The is error will be discarded.");
	    return;
	}

	jobStatus.setRunning(false);
	jobStatus.setErrorCount(jobStatus.getErrorCount() + 1);
	if (jobStatus.getErrorCount() > this.jobErrorLimit) {
	    this.logger.warn("Job#" + job.getId() + " report error more than limits, will be reported fail.");
	    this.jobStatuses.remove(job.getId());
	    handleJobFailure(job);
	}
    }

    @Override
    public synchronized void handleJobResult(JobResult result) throws JobDispatcherException {
	JobStatus record = this.jobStatuses.get(result.getId());
	if (record == null) {
	    this.logger.warn("The status record of job#" + result.getId()
		    + " has already been removed. The job result will be discarded.");
	    return;
	}

	record.setFinished(true);

	this.jobResultCache.add(result);
	if (this.jobResultCache.size() >= this.flushThreashold || !hasMore()) {
	    try {
		// begin flush thread
		JobResultFlusher flusher = new JobResultFlusher(this.jobResultCache);
		if (this.useSeparateThreadFlush) {
		    new Thread(flusher).start();
		} else {
		    flusher.run();
		}
	    } finally {
		// clear job result cache no matter what happened during flush
		this.jobResultCache.clear();
	    }
	}
    }

    @Override
    public synchronized boolean hasMore() throws JobDispatcherException {
	for (JobStatus jobStatus : this.jobStatuses.values()) {
	    if (!jobStatus.isFinished()) {
		return true;
	    }
	}

	return false;
    }

    protected abstract List<Job> retrieveNewJobs() throws JobDispatcherException;

    protected abstract void flushJobResult(List<JobResult> jobResults) throws JobDispatcherException;

    protected abstract void handleJobFailure(Job job) throws JobDispatcherException;

    private void findJobsFromMemory(List<Job> jobs, int jobCount) {
	for (JobStatus jobStatus : new ArrayList<JobStatus>(this.jobStatuses.values())) {
	    if (jobStatus.getLastDispatchTime() != null
		    && new Date().getTime() - jobStatus.getLastDispatchTime().getTime() > this.jobRunningTimeout) {
		this.logger.warn("Job#" + jobStatus.getJob().getId() + " running timeout, will be reported fail.");
		this.jobStatuses.remove(jobStatus.getJob().getId());
		handleJobFailure(jobStatus.getJob());
		continue;
	    }

	    if (!jobStatus.isFinished() && !jobStatus.isRunning()) {
		jobs.add(jobStatus.getJob());

		jobStatus.setRunning(true);
		jobStatus.setLastDispatchTime(new Date());

		if (jobs.size() == jobCount) {
		    return;
		}
	    }
	}
    }

    protected class JobStatus {
	private Job job;
	private boolean running;
	private Date lastDispatchTime;
	private boolean finished;
	private int errorCount;

	public void setRunning(boolean crawling) {
	    this.running = crawling;
	}

	public boolean isRunning() {
	    return running;
	}

	public void setLastDispatchTime(Date lastDispatchTime) {
	    this.lastDispatchTime = lastDispatchTime;
	}

	public Date getLastDispatchTime() {
	    return lastDispatchTime;
	}

	public void setFinished(boolean finished) {
	    this.finished = finished;
	}

	public boolean isFinished() {
	    return finished;
	}

	public void setErrorCount(int failCount) {
	    this.errorCount = failCount;
	}

	public int getErrorCount() {
	    return errorCount;
	}

	public void setJob(Job job) {
	    this.job = job;
	}

	public Job getJob() {
	    return job;
	}
    }

    public int getFlushThreashold() {
	return flushThreashold;
    }

    public void setFlushThreashold(int flushThreashold) {
	this.flushThreashold = flushThreashold;
    }

    public int getJobRunningTimeout() {
	return jobRunningTimeout;
    }

    public void setJobRunningTimeout(int jobRunningTimeout) {
	this.jobRunningTimeout = jobRunningTimeout;
    }

    public int getJobErrorLimit() {
	return jobErrorLimit;
    }

    public void setJobErrorLimit(int jobErrorLimit) {
	this.jobErrorLimit = jobErrorLimit;
    }

    public void setUseSeparateThreadFlush(boolean useSeparateThreadFlush) {
	this.useSeparateThreadFlush = useSeparateThreadFlush;
    }

    public boolean isUseSeparateThreadFlush() {
	return useSeparateThreadFlush;
    }

    private class JobResultFlusher implements Runnable {
	private List<JobResult> jobResults;

	public JobResultFlusher(List<JobResult> jobResults) {
	    this.jobResults = new ArrayList<JobResult>(jobResults);
	}

	@Override
	public void run() {
	    try {
		long startTime = new Date().getTime();
		AbstractDispatcher.this.logger.info("Start flushing job results...");
		flushJobResult(this.jobResults);
		AbstractDispatcher.this.logger.info(jobResults.size() + " job results flushed, cost "
		        + ((new Date().getTime() - startTime) / 1000.0) + "s.");
	    } finally {
		// clear job status not matter what happened during flush
		synchronized (AbstractDispatcher.this) {
		    for (JobResult jobResult : this.jobResults) {
			AbstractDispatcher.this.jobStatuses.remove(jobResult.getId());
		    }
		    AbstractDispatcher.this.logger.info("Job statuses size: " + jobStatuses.size());
		}
	    }
	}
    }
}
